package com.my.study.batch;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 请求合并器
 *
 * @author Carlos
 * @version 1.0.0
 * @date 2023/3/24 21:48
 */
@Slf4j(topic = "c.BatchCollapsed")
public class BatchCollapsed {

    private static final Integer CAPACITY = 4;

    private volatile LinkedBlockingDeque<Work> batchContainer = new LinkedBlockingDeque<>(CAPACITY);

    private static final ScheduledExecutorService SCHEDULED =
            Executors.newScheduledThreadPool(1, r -> {
                Thread schedule = new Thread(r, "schedule");
                schedule.setDaemon(true);
                return schedule;
            });

    public BatchCollapsed() {
        schedule();
    }

    private void schedule() {
        SCHEDULED.scheduleAtFixedRate(() -> {
            try {
                deal();
            } catch (Exception e) {
                log.error("deal error", e);
            }
        }, 1000, 1000, TimeUnit.MILLISECONDS);
        
    }

    public void submit(Work work) {
        for (; ; ) {
            boolean putSuccess = batchContainer.offerLast(work);
            if (putSuccess) {
                break;
            }
        }
        if (batchContainer.size() >= CAPACITY) {
            // 批量处理请求
            deal();
        }
    }


    private void deal() {
        List<Work> works = new ArrayList<>(CAPACITY);
        int drain = batchContainer.drainTo(works, CAPACITY);
        log.debug("[deal] drain: {}, size: {}", drain, works.size());
    }

    public static void main(String[] args) throws InterruptedException {
        BatchCollapsed batchCollapsed = new BatchCollapsed();

        AtomicInteger no = new AtomicInteger(1);
        while (no.get() <= 10) {
            final String seq = no.getAndIncrement() + "";
            log.debug("seq: {}", seq);
            new Thread(() -> batchCollapsed.submit(new Work(seq)), seq).start();
        }

        TimeUnit.SECONDS.sleep(3);
        List<Work> works = new ArrayList<>();
        int remain = batchCollapsed.batchContainer.drainTo(works);
        log.debug("剩余 {}", remain);
    }

}